package com.more.offer.threadtest.threadpool;

import com.google.common.util.concurrent.ThreadFactoryBuilder;

import java.util.concurrent.*;

/**
 * @author pinoc
 * @date 2021/7/6
 * @description 模拟binlog 消息同步 线程池处理
 * businessThread ：业务处理（监听同步）
 * warningThread: 监控告警，监控告警内部发送邮件或者消息（发送逻辑占用时常约2s-5s），因此告警不能与businessThread同步执行
 * 异步告警处理，达到最大线程数其余任务丢弃，业务处理，自定义策略使用阻塞队列，知道线程池有空闲线程处理任务继续处理。
 */
public class ThreadPoolTest {

	public static void main(String[] args) {

		ThreadPoolExecutor businessExecutor = new ThreadPoolExecutor(2, 6, 10,
				TimeUnit.MICROSECONDS,
				new SynchronousQueue<>(true),
				new ThreadFactoryBuilder().setNameFormat("business-thread-poll-thread").build(),
				ThreadPoolTest.rejected_handler);

		ThreadPoolExecutor warningExecutor = new ThreadPoolExecutor(2, 2, 10,
				TimeUnit.MICROSECONDS,
				new LinkedBlockingDeque<Runnable>(20),
				new ThreadFactoryBuilder().setNameFormat("warning-thread-poll-thread").build(),
				new ThreadPoolExecutor.DiscardPolicy());

		for (int i = 0; i < 201; i++) {
			final int index = i;
			businessExecutor.submit(() -> {
				System.out.println("业务处理start..." + index);
					warningExecutor.submit(new MyThread("warning-thrad-" + index));
				System.out.println("业务处理end..."+ index);
			});
		}

		System.out.println("业务处理完成。。。。");
	}

	static class MyThread implements Runnable {
		String name;
		public MyThread(String name) {
			this.name = name;
		}
		@Override
		public void run() {
			try {
				Thread.sleep(1000);
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
			System.out.println(this.name+"-告警");
		}
	}

	/**
	 * 自定义拒绝策略
	 */
	public static final RejectedExecutionHandler rejected_handler = (r, executor) -> {
		if(r!=null) {
			try {
				executor.getQueue().put(r);
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
		}
	};
}
